#ifndef _MESSAGEUTIL_H_
#define _MESSAGEUTIL_H_

#include "../inc/package.h"
#include "command_config.h"
#include "../inc/logger.h"
#include "zlib.h"
#include "google/protobuf/io/coded_stream.h"
#include <stdio.h>
#include "json2pb.h"

#define MAXRATIO 1024

using namespace google::protobuf;

class MessageUtil
{
public:
	static bool serializePackageToString(PackageHeader& header,const char* data,unsigned int size,bool zip,unsigned int zipThreshold,std::string* output)
	{
		string out;
		if(header.protocol()==PackageHeader::CodeType::PROTOBUF && zip && size>zipThreshold)
		{
			unsigned int ratio;
			bool ret = MessageUtil::Compress(data,size,out,ratio);
			if(ret == false/* || out.size() >= size*/)
			{
				header.set_iszip(0);
				header.set_compratio(0);
			}
			else if(out.size() >= size)
			{
				header.set_iszip(1);
				header.set_compratio(1);
			}
			else if(ratio > 1024)
			{
				header.set_iszip(1);
				header.set_compratio(0);
			}
			else
			{
				header.set_iszip(1);
				header.set_compratio ((unsigned short)ratio);
			}
		}
		else
		{
			header.set_iszip(0);
			header.set_compratio(0);
		}
		
		output->clear();
		
		if(header.iszip())
		{
			header.set_bodysize(out.size());
			output->reserve(PackageHeaderLength+header.bodysize());
		    output->resize(PackageHeaderLength);
		    header.write(&*output->begin());
		    output->append(out);
		}
		else
		{
			header.set_bodysize(size);
			output->reserve(PackageHeaderLength+size);
		    output->resize(PackageHeaderLength);
		    header.write(&*output->begin());
			output->append(data,size);
		}
		return true;
	}
	static bool serializePackageToString(PackageHeader& header,const Message& msg,bool zip,unsigned int zipThreshold,std::string* output)
	{
		std::string data;
		if (header.protocol() == PackageHeader::CodeType::PROTOBUF)
		{
			try
			{
				if (!msg.AppendToString(&data))
					return false;
			}
			catch (FatalException err)
			{
				LOG4_ERROR("Serialize error: %s", err.message().c_str());
				return false;
			}
			return serializePackageToString(header, data.data(), (unsigned int)(data.size()), zip, zipThreshold, output);
		}
		else if (header.protocol() == PackageHeader::CodeType::JSON)
		{
			data=pb2json(msg);
			return serializePackageToString(header, data.data(), (unsigned int)(data.size()), false, 0, output);
		}
		else
		{
			LOG4_ERROR("CodeType not support");
			return false;
		}
		
	}
	static bool serializeExtMessageToString(const Message& extMsg,std::string* output)
	{
		std::string typeName=extMsg.GetTypeName();
		size_t oldSize=output->size();
		output->resize(oldSize+3);
		output->at(oldSize+2)=(unsigned char)typeName.size();
		output->append(typeName);
		try
		{
		    if(!extMsg.AppendToString(output))
			    return false;
		}
		catch(FatalException err)
		{
			LOG4_ERROR("Serialize error: %s",err.message().c_str());
			return false;
		}
		*((unsigned short*)output->data()+oldSize)=htons(output->size());
		return true;
	}

	static Message* createMessage(string clazz)
	{
		Message* message = NULL;
		do{
			const Descriptor* des=DescriptorPool::generated_pool()->FindMessageTypeByName(clazz);
			if(!des)
				break;	
			const Message* prototype = MessageFactory::generated_factory()->GetPrototype(des);
			if (!prototype)
				break;
			message = prototype->New();
		}while(false);
		return message;
	}
	static bool parseFromArray(Message* message,const char* data,size_t size,bool zipped,unsigned char compratio)
	{
		google::protobuf::io::CodedInputStream* stream=NULL;
		string* msgcontent=NULL;
		bool ret=false;
		do{
			if(zipped)
			{
				msgcontent=new string();
			    bool unzipOK = MessageUtil::unCompress(data,size,compratio,*msgcontent);
			    if(!unzipOK)
					break;
				stream=new google::protobuf::io::CodedInputStream((const uint8*)msgcontent->data(),msgcontent->size());
			}
			else
			    stream=new google::protobuf::io::CodedInputStream((const uint8*)data,size);
			if(message->ParseFromCodedStream(stream))
				ret=true;
			else
				LOG4_ERROR("ParseFromCodedStream failed");
		}while(false);
		if(msgcontent)
			delete msgcontent;
		if(stream)
			delete stream;
		return ret;
	}
	static Message* parseFromArray(string clazz,const char* data,size_t size,bool zipped,unsigned char compratio)
	{
		Message* message = createMessage(clazz);
		if(message && !parseFromArray(message,data,size,zipped,compratio))
		{
			delete message;
			message=NULL;
		}
		return message;
	}
	static bool parse(Package* package,string& retstr)
	{
		const char* bodyData = package->GetBodyData();
		unsigned int bodySize = package->GetBodySize();
		string data;
		if (package->header().iszip())
		{
			bool unzipOK = MessageUtil::unCompress(bodyData, bodySize, package->header().compratio(), data);
			if (!unzipOK)
			{
				retstr = "unCompress failed";
				LOG4_ERROR("unCompress failed");
				return false;
			}
			bodyData = data.data();
			bodySize = data.size();
		}
		if (package->message() == nullptr)
		{
			Message* message = createMessage(package->classtype());
			if (!message)
			{
				retstr = "create message failed:" + package->classtype();
				LOG4_ERROR("create message failed,type=%s", package->classtype().c_str());
				return false;
			}
			package->set_message(MessagePtr(message));
		}
		unsigned short extSize = 0;
		if (package->header().hasext())
		{
			extSize = *((unsigned short*)bodyData);
			extSize = ntohs(extSize);
			unsigned char nameSize = *(unsigned char*)(bodyData + 2);
			string extClazz;
			extClazz.assign(bodyData + 3, nameSize);
			Message* extMsg = parseFromArray(extClazz, bodyData + nameSize + 3, extSize - nameSize - 3, false, 0);
			package->set_extmessage(MessagePtr(extMsg));
		}
		if (package->header().protocol() == PackageHeader::CodeType::PROTOBUF)
			return parseFromArray(package->message().get(), bodyData + extSize, bodySize - extSize, false, 0);
		else if (package->header().protocol() == PackageHeader::CodeType::JSON)
		{
			try
			{
				json2pb(*package->message().get(), bodyData + extSize, bodySize - extSize);
			}
			catch (std::exception& err)
			{
				retstr = "json2pb failed:" + string(err.what());
				LOGGER_ERROR("json2pb failed:" << err.what());
				return false;
			}
			return true;
		}
		return false;
	}
	static bool parse(Package* package)
	{
		string str;
		return parse(package, str);
	}
	static bool Compress(const char* data,unsigned long size,string& out,unsigned int &ratio)
	{
		uLong uncomlen=(uLong)size;
		uLong destSize=compressBound(uncomlen);
		size_t old=out.size();
		out.resize(old+destSize);
		int ret =compress((Bytef*)(out.data()+old), &destSize, (const Bytef*)data,uncomlen);
		if(Z_OK!=ret)
		{
			LOG4_ERROR("compress error: %d",ret);
			return false;
		}
		ratio= (unsigned int)(uncomlen /destSize + 1);
		out.resize(old+destSize);
		return true;
	}

	static bool  unCompress(const char * data,size_t size,unsigned short ratio,string &out)	
	{
		uLong comprLen=(uLong)size;
		uLong uncomprLen=ratio != 0  ? ratio*comprLen + 1 : MAXRATIO*comprLen;  
		size_t old=out.size();
		out.resize(old+uncomprLen);
		if(out.capacity()<uncomprLen)
			return false;	
		int ret=uncompress((Bytef*)(out.data()+old),&uncomprLen,(const Bytef*)(data),comprLen);
		if(Z_OK!=ret)
		{
			LOG4_ERROR("uncompress error: %d",ret);
			return false;
		}
		out.resize(old+uncomprLen);
		return true;
	}
};

#endif